package mapreduce;

import java.io.File;
import java.util.Collections;
import java.util.Iterator;
import java.util.Vector;

public class Task {
	public Task(JobConf conf, int id) {
		m_conf = conf;
		m_id = id;
		initial();
	}

	public void run() {
		m_timer = new MyTimer();
		RecordFeeder feeder = m_conf.getFeeder();
		RecordWriter writer = m_conf.getWriter();
		Mapper mapper = m_conf.getMapper();
		Reducer reducer = m_conf.getReducer();

		System.out.println("mapping......");
		map(mapper, feeder);
		double mapTime = 0, sortTime = 0, reduceTime = 0;
		mapTime = m_timer.now() * .001;
		if (reducer != null) {
			System.out.println("sorting......");
			sort();
			sortTime = m_timer.now() * .001 - mapTime;
			System.out.println("reducing......");
			reduce(reducer, writer);
			reduceTime = m_timer.now() * .001 - sortTime;
			release();
		}
		System.out.println(toString() + ":"
				+ new Double(m_timer.now() * 0.001).toString() + "("
				+ new Double(mapTime).toString() + ","
				+ new Double(sortTime).toString() + ","
				+ new Double(reduceTime).toString() + ")");
	}

	private void initial() {
		String idstr = new Integer(m_id).toString();
		m_mapResultPath = "map.result." + idstr;
		m_sortResultPath = "sort.result." + idstr;
	}

	private void map(Mapper mapper, RecordFeeder feeder) {
		int count = 0;
		MyTimer timer = new MyTimer();
		RecordWriter mapResultWriter;
		if (m_conf.getReducer() != null) {
			mapResultWriter = new RecordWriter();
			mapResultWriter.open(new File(m_mapResultPath));
		} else {
			mapResultWriter = m_conf.getWriter();
		}
		Vector<Record> outRecords = new Vector<Record>();
		while (true) {
			int next = feeder.next();
			if (next == -1) {
				break;
			}
			else if(next == -2) {
				continue;
			}
			count++;
			if (count % 100000 == 0) {
				System.out.println(new Integer(count).toString() + " "
						+ new Integer((int) (timer.now() * 0.001)).toString());
			}

			Record inRecord = feeder.getRecord();
			mapper.map(inRecord, outRecords);
			writeResult(mapResultWriter, outRecords);
			outRecords.clear();

		}
		mapper.release(outRecords);
		writeResult(mapResultWriter, outRecords);
		outRecords.clear();
		mapResultWriter.close();
	}

	private void writeResult(RecordWriter writer, Vector<Record> records) {
		Iterator<Record> recordIt = records.iterator();
		while (recordIt.hasNext()) {
			Record record = recordIt.next();
			writer.write(record);
		}
	}

	private int partSort(Vector<Record> records, int partNo) {
		Collections.sort(records, new RecordDefaultComparator());
		RecordWriter writer = new RecordWriter();
		writer.open(new File(m_sortResultPath + "." + new Integer(partNo).toString()));
		writeResult(writer, records);
		writer.close();

		return partNo + 1;
	}

	private void sort() {
		MyTimer timer = new MyTimer();
		int partNum = 0;
		RecordReader mapResultReader = new RecordReader(new File(m_mapResultPath));
		RecordDefaultComparator recordComp = new RecordDefaultComparator();
		// 1. part sort
		Vector<Record> currRecords = new Vector<Record>();
		int currRecordCnt = 1;
		long totalCnt = 1;
		mapResultReader.next();
		Record record = mapResultReader.getRecord();
		int partSize = m_conf.getPartSize();
		System.out.println("part size:" + new Long(partSize).toString());
		while (true) {
			currRecords.add(record);
			if (currRecordCnt >= partSize) {
				System.out.println("part sorting "
						+ new Long(totalCnt).toString() + " "
						+ new Integer((int) (timer.now() * .001)).toString());
				partNum = partSort(currRecords, partNum);
				currRecords = new Vector<Record>();
				Runtime.getRuntime().gc();
				currRecordCnt = 0;
			}
			if (mapResultReader.next() == -1) {
				break;
			}
			record = mapResultReader.getRecord();
			// byte[] bytes = record.getReadableBytes();
			// System.out.println(new String(bytes));
			currRecordCnt++;
			totalCnt++;
			if (totalCnt % 200000 == 0) {
				System.out.println(new Long(totalCnt).toString() + " "
						+ new Integer((int) (timer.now() * .001)).toString());
			}
		}
		if (currRecordCnt != 0) {
			partNum = partSort(currRecords, partNum);
		}

		// 2. merge sort
		totalCnt = 0;
		System.out.println("mergering......");
		RecordReader[] partSortResultReaders = new RecordReader[partNum];
		RecordWriter sortRecordWriter = new RecordWriter();
		sortRecordWriter.open(new File(m_sortResultPath));
		Record[] records = new Record[partNum];
		for (int i = 0; i < partNum; i++) {
			partSortResultReaders[i] = new RecordReader(new File(m_sortResultPath + "."
					+ new Integer(i).toString()));
			partSortResultReaders[i].next();
			records[i] = partSortResultReaders[i].getRecord();
		}
		while (true) {
			boolean allEmpty = (records[0] == null);
			int minIndex = 0;
			for (int i = 1; i < partNum; i++) {
				if (records[i] != null) {
					allEmpty = false;
				}
				int comp = recordComp.compare(records[minIndex], records[i]);
				if (comp > 0) {
					minIndex = i;
				}
			}
			if (allEmpty) {
				break;
			}
			sortRecordWriter.write(records[minIndex]);
			partSortResultReaders[minIndex].next();
			records[minIndex] = partSortResultReaders[minIndex].getRecord();
			totalCnt++;
			if (totalCnt % 200000 == 0) {
				System.out.println(new Long(totalCnt).toString() + " "
						+ new Integer((int) (timer.now() * .001)).toString());
			}
		}
		for (int i = 0; i < partNum; i++) {
			partSortResultReaders[i].close();
			File sortFile = new File(m_sortResultPath + "."
					+ new Integer(i).toString());
			sortFile.delete();
		}
		sortRecordWriter.close();

	}

	private void reduce(Reducer reducer, RecordWriter writer) {
		RecordDefaultComparator recordComp = new RecordDefaultComparator();
		RecordReader sortRecordReader = new RecordReader(new File(m_sortResultPath));

		MyTimer timer = new MyTimer();
		int count = 0;
		sortRecordReader.next();
		Vector<Record> records = new Vector<Record>();
		Record record = sortRecordReader.getRecord();
		records.add(record);
		Record prevRecord = record;
		while (sortRecordReader.next() != -1) {
			record = sortRecordReader.getRecord();
			if (recordComp.compareKey(record, prevRecord) == 0) {
				records.add(record);
			} else {
				partReduce(reducer, writer, records);
				records.clear();
				records.add(record);
				prevRecord = record;
			}
			count++;
			if (count % 200000 == 0) {
				System.out.println(new Integer(count).toString() + " "
						+ new Integer((int) (timer.now() * .001)).toString());
			}
		}
		partReduce(reducer, writer, records);
		Vector<Record> outRecords = new Vector<Record>();
		reducer.release(outRecords);
		writeResult(writer, outRecords);

		writer.close();
		sortRecordReader.close();
	}

	private void partReduce(Reducer reducer, RecordWriter writer,
			Vector<Record> inRecords) {
		Vector<Record> outRecords = new Vector<Record>();
		reducer.reduce(inRecords, outRecords);
		writeResult(writer, outRecords);
	}

	private void release() {
		File mapResultFile = new File(m_mapResultPath);
		File sortResultFile = new File(m_sortResultPath);
		mapResultFile.delete();
		sortResultFile.delete();
	}

	JobConf m_conf;
	int m_id;

	String m_mapResultPath;
	String m_sortResultPath;
	MyTimer m_timer;
}
